package dag

import (
	"fmt"
	"github.com/shiningrush/fastflow"
	mongoKeeper "github.com/shiningrush/fastflow/keeper/mongo"
	"github.com/shiningrush/fastflow/pkg/entity/run"
	mongoStore "github.com/shiningrush/fastflow/store/mongo"
	"hxy/repo/database"
	"log"
)

func init() {
	initDagDB()
	database.Db.AutoMigrate(&DagTask{})
}

func initDagDB() {
	fastflow.RegisterAction([]run.Action{
		&DAGPipelineTask{},
	})

	// init keeper, it used to e
	keeper := mongoKeeper.NewKeeper(&mongoKeeper.KeeperOption{
		Key: "worker-1",
		// if your mongo does not set user/pwd, you should remove it
		ConnStr:  "mongodb://root:qwerasdf@192.168.137.1:27017/fastflow?authSource=admin",
		Database: "mongo-demo",
		Prefix:   "test",
	})
	if err := keeper.Init(); err != nil {
		log.Println(fmt.Errorf("init keeper failed: %w", err))
	}

	// init store
	st := mongoStore.NewStore(&mongoStore.StoreOption{
		// if your mongo does not set user/pwd, you should remove it
		ConnStr:  "mongodb://root:qwerasdf@192.168.137.1:27017/fastflow?authSource=admin",
		Database: "mongo-demo",
		Prefix:   "test",
	})
	if err := st.Init(); err != nil {
		log.Println(fmt.Errorf("init store failed: %w", err))
	}

	// start fastflow
	if err := fastflow.Init(&fastflow.InitialOption{
		Keeper: keeper,
		Store:  st,
		// use yaml to define dag
		ReadDagFromDir: "./",
	}); err != nil {
		log.Println(fmt.Errorf("init fastflow failed: %s", err))
	}
}
